package com.ibm.cps.model;

import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

import org.apache.log4j.Logger;
import org.python.google.common.base.Throwables;

import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.ibm.cps.message.MessageFields;
import com.ibm.cps.message.MetadataTopologySort;
import com.ibm.cps.newmessage.AbstractMetadata;
import com.ibm.cps.newmessage.MetadataFactory;
import com.ibm.cps.processors.AbstractProcessor;
import com.ibm.cps.processors.ProcessorFactory;
import com.ibm.cps.processors.ProcessorsGroup;
import com.ibm.interfaces.IPersistency;
import com.ibm.mongo.MongoPersistency;
import com.ibm.util.ErrorCode;
import com.ibm.util.JSONUtility;
import com.ibm.util.exception.CPSException;

public class ProcessorMetadtaValidator {

	private static Logger logger;
	private static ProcessorFactory factory;
	private static HashMap<String, AbstractProcessor> id2Processor;
	private static HashSet<String> existedProcessorid;
	private static IPersistency persistency;
	private static HashMap<String, Set<String>> processor2successors;
	private static HashSet<String> existedGroups;
	private static ProcessorMetadtaValidator validator = null;
	private static boolean isFirstMetadataLoad = true;
	private static boolean isFirstGroupLoad = true;

	public static ProcessorMetadtaValidator getInstance() throws CPSException {
		if (validator == null) {
			synchronized ("lock") {
				if (validator == null) {
					validator = new ProcessorMetadtaValidator();
				}
			}
		}
		return validator;
	}

	private ProcessorMetadtaValidator() throws CPSException {
		logger = Logger.getLogger(ProcessorMetadtaValidator.class);
		factory = ProcessorFactory.getInstance();
		id2Processor = new HashMap<String, AbstractProcessor>();
		persistency = MongoPersistency.getInstace();
		existedProcessorid = new HashSet<String>();
		processor2successors = new HashMap<String, Set<String>>();
		existedGroups = new HashSet<String>();
	}

	private static void loadExistingMetadata() {
		try {
			if (isFirstMetadataLoad) {
				Collection<String> metadatas = persistency.getProcessors(null, null, null, null);
				if (metadatas != null) {

					List<ObjectNode> sortedMetadatas = MetadataTopologySort.getSortedMetadatas(metadatas);

					for (ObjectNode message1 : sortedMetadatas) {
						try {
							AbstractMetadata metadata = MetadataFactory.parseJsonMetadata(message1.toString());
							MetadataFactory.updateProcessorMap(processor2successors, metadata);
							validateMetadata(metadata.getProcessorid(), metadata);
						} catch (CPSException e) {
							logger.warn("Error processor metadata stored in mongo: " + message1.toString());
						}
					}
				}
				isFirstMetadataLoad = false;
			}
		} catch (CPSException e) {
			logger.error(Throwables.getStackTraceAsString(e));
		}
	}

	private static void loadExistingGroup() {
		try {
			if (isFirstGroupLoad) {
				Collection<String> groups = persistency.getAllGroupMetadatas();
				if (groups != null) {
					for (String groupInfo : groups) {

						if (groupInfo != null) {
							ObjectNode node = (ObjectNode) JSONUtility.fromString(groupInfo);
							JSONUtility.checkNotNull(node, MessageFields.GROUPID);
							String groupid = node.get(MessageFields.GROUPID).asText();
							existedGroups.add(groupid);
						}

					}
				}
				existedGroups.add(ProcessorsGroup.DEFAULT_GROUPTID);
				isFirstGroupLoad = false;
			}
		} catch (CPSException e) {
			logger.error("Error group metadata stored in mongo.");
			logger.error(Throwables.getStackTraceAsString(e));
		}
	}

	public boolean validate(String tenantid, String processorid, ObjectNode node, String source) throws CPSException {
		firstLoad();

		// all processors should belong to a group
		String type = node.get(MessageFields.TYPE).asText();
		if (ProcessorsGroup.isGroupType(type)) {
			if (existedGroups.contains(processorid)) {
				throw new CPSException(ErrorCode.ALREADY_EXIST_CODE, "Group " + processorid + " already existed.");
			}
			existedGroups.add(processorid);
			return true;
		}

		AbstractMetadata metadata = MetadataFactory.parseJsonMetadata(node);

		String groupid = node.get(MessageFields.GROUPID).asText();
		if (!existedGroups.contains(groupid)) {
			throw new CPSException(ErrorCode.INVALID_INPUT_ERROR, "Group " + groupid + " doesn't exist");
		}
		if (id2Processor.containsKey(processorid)) {
			AbstractProcessor processor = id2Processor.get(processorid);
			String initialSource = processor.getSource();

			// if node-red trys to update message of sql, do nothing
			if (!initialSource.equals(source)) {
				return false;
			}
		}
		MetadataFactory.updateProcessorMap(processor2successors, metadata);
		validateMetadata(processorid, metadata);
		if (!metadata.getTenantid().equals(tenantid)) {
			throw new CPSException(ErrorCode.CAN_NOT_UPDATE_CODE, "Can't change the tenantid.");
		}
		return true;
	}

	private void firstLoad() throws CPSException {
		if (isFirstMetadataLoad) {
			loadExistingMetadata();
		}
		if (isFirstGroupLoad) {
			loadExistingGroup();
		}
	}

	private static void validateMetadata(String processorid, AbstractMetadata metadata) throws CPSException {

		if (id2Processor.containsKey(processorid)) {
			throw new CPSException(ErrorCode.ALREADY_EXIST_CODE, "Processor " + processorid + " already existed.");
		}
		AbstractProcessor processor = factory.createProcessor(metadata, id2Processor);
		factory.updateProcessorMap(metadata.getProcessorid(), processor, id2Processor, metadata);
		existedProcessorid.add(processorid);
	}

	public ArrayNode deleteProcessor(String tenantid, String deletedMessage, String source) throws CPSException {
		firstLoad();
		ArrayNode array = JSONUtility.newArrayNode();

		if (deleteGroupMetadata(tenantid, deletedMessage, source)) {
			return array;
		}

		AbstractMetadata metadata = MetadataFactory.parseJsonMetadata(deletedMessage);
		if (existedProcessorid.contains(metadata.getProcessorid())){
//		if (existedProcessorid.contains(metadata.getProcessorid())
//				&& (source == null || (source.equals("rest") && !metadata.getSource().equals(MessageFields.GROUP_TYPE_SQL)
//						|| source.equals(metadata.getSource())))) {
			persistency.deleteProcessors(tenantid, metadata.getProcessorid(), null, null);
			id2Processor.remove(metadata.getProcessorid());
			existedProcessorid.remove(metadata.getProcessorid());
			array.add(metadata.getProcessorid());
		} else {
			return null;
		}

		List<List<Object>> retMsg = MetadataFactory.deleteProcessor(processor2successors, metadata);

		String deletedProcessor;
		if (retMsg != null && retMsg.size() >= 1) {
			for (List<Object> delObject : retMsg) {
				deletedProcessor = (String) delObject.get(1);
				if (existedProcessorid.contains(deletedProcessor)
						&& (source == null || (source.equals("rest") && !metadata.getSource().equals(MessageFields.GROUP_TYPE_SQL))
								|| metadata.getSource().equals(source))) {
					array.add(deletedProcessor);
					id2Processor.remove(deletedProcessor);
					persistency.deleteProcessors(tenantid, deletedProcessor, null, null);
					// persistency.deleteMetadataByUserProcessor(tenantid,
					// deletedProcessor);
					existedProcessorid.remove(deletedProcessor);
				}
			}
		}
		return array;
	}

	private boolean deleteGroupMetadata(String tenantid, String deletedMessage, String source) throws CPSException {
		ObjectNode objectNode = JSONUtility.fromObjectString(deletedMessage);
		String type = objectNode.get(MessageFields.TYPE).asText();
		if (ProcessorsGroup.isGroupType(type)) {
			String groupid = objectNode.get(MessageFields.GROUPID).asText();
			persistency.deleteGroupMetadata(tenantid, groupid);
			// persistency.deleteGroupById(tenantid, groupid);
			existedGroups.remove(groupid);
			return true;
		} else if (ProcessorsGroup.isGroupType(type)) {
			return true;
		}
		return false;
	}
}
